package org.lzfto.flink.demo.business.flinkTest.trigger;

import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 计数+超时复合触发器 <br/>
 * 如果把 W 替换成通用类型，则这是一个通用的触发器
 * 
 * @author lzfto
 * @创建时间 2025年6月20日 17:36:38
 */
public class CountOrTimeTrigger<T,W extends Window> extends Trigger<T, W> {
	private static final long serialVersionUID = 1L;

	private static final Logger LOGGER = LoggerFactory.getLogger(CountOrTimeTrigger.class);

	private final String name;
	private final long maxCount;
	private final long intervalMillis;

	private final ReducingStateDescriptor<Long> countStateDesc = new ReducingStateDescriptor<>("count", Long::sum,
			Long.class);

	private final ValueStateDescriptor<Long> timerStateDesc = new ValueStateDescriptor<>("timer", Long.class);

	
	public CountOrTimeTrigger(long maxCount, long intervalMillis) {
		this.name="计数或者计时触发器";
		this.maxCount = maxCount;
		this.intervalMillis = intervalMillis;
		LOGGER.info("创建触发器:{}-每{}条记录触发，或者每{}毫秒周期触发",name,maxCount,intervalMillis);
	}
	
	public CountOrTimeTrigger(String name,long maxCount, long intervalMillis) {
		this.name=name;
		this.maxCount = maxCount;
		this.intervalMillis = intervalMillis;
		LOGGER.info("创建触发器:{}-每{}条记录触发，或者每{}毫秒周期触发",name,maxCount,intervalMillis);
	}

	@Override
	public TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception {
		ReducingState<Long> count = ctx.getPartitionedState(countStateDesc);
		ValueState<Long> timer = ctx.getPartitionedState(timerStateDesc);
		count.add(1L);
		boolean matchFireCondition=false;
		if (count.get() == maxCount) {
			count.clear();			
			matchFireCondition=true;
		}

		if (timer.value()!=null) {
			ctx.deleteProcessingTimeTimer(timer.value());
		}
		
		//重置触发器
		long newTimer=ctx.getCurrentProcessingTime()+intervalMillis;
		timer.update(newTimer);
		ctx.registerProcessingTimeTimer(newTimer);
		
		if (matchFireCondition) {
			return TriggerResult.FIRE;
		}
		else {
			return TriggerResult.CONTINUE;
		}
	}

	@Override
	public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
		return TriggerResult.CONTINUE;
	}

	@Override
	public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
		ReducingState<Long> count = ctx.getPartitionedState(countStateDesc);
		ValueState<Long> timer = ctx.getPartitionedState(timerStateDesc);
		
		boolean matchFireCondition=false;
		Long cqty=count.get();
		if(cqty!=null && cqty>0 ) {
			matchFireCondition=true;
		}
		
		if (matchFireCondition) {
			long nextTime = ctx.getCurrentProcessingTime() + intervalMillis;
			ctx.registerProcessingTimeTimer(nextTime);
			timer.update(nextTime);
			count.clear();
			return TriggerResult.FIRE;
		}
		else {
			return TriggerResult.CONTINUE;
		}
	}

	@Override
	public void clear(W window, TriggerContext ctx) throws Exception {
		ctx.getPartitionedState(countStateDesc).clear();
		ctx.getPartitionedState(timerStateDesc).clear();
		LOGGER.warn("触发器环境已经清除自定义状态:开始时间、最后运行时间");
	}

}
